package com.dyrnq.seckill.mq.hazelcast;


import com.dyrnq.seckill.mq.MessageVo;
import com.dyrnq.seckill.mq.Producer;
import com.dyrnq.seckill.service.impl.SeckillOrderService;
import com.hazelcast.collection.IQueue;
import com.hazelcast.core.HazelcastInstance;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;


@Component
@ConditionalOnProperty(value = "queue.impl", havingValue = "hazelcast")
@RequiredArgsConstructor
@Slf4j
public class HazelcastConsumer implements ApplicationRunner {
    private final HazelcastInstance hazelcastInstance;

    private final SeckillOrderService seckillOrderService;

    @Override
    public void run(ApplicationArguments args) throws Exception {

//        Ringbuffer<MessageVo> rb = hazelcastInstance.getRingbuffer(Producer.QUEUE_NAME);
//        long sequence = rb.headSequence();

        IQueue<MessageVo> queue = hazelcastInstance.getQueue(Producer.QUEUE_NAME);
        while (true) {
            MessageVo msg = queue.take();
            if (msg != null) {
                log.info("Message GoodsId {} , UserId {}", msg.getGoodsId(), msg.getUserId());
                long goods_id = msg.getGoodsId();
                long user_id = msg.getUserId();
                seckillOrderService.insert(goods_id, user_id);
            }
        }
    }

}
